package com.business.leaf.core.segment.domain.aggregate;

import com.business.leaf.core.segment.domain.entity.SegmentBuffer;
import com.business.leaf.core.segment.domain.valobj.Segment;
import com.business.leaf.core.segment.ex.LeafEx;
import com.business.leaf.core.segment.repo.mapper.LeafAllocMapper;
import com.business.leaf.core.segment.repo.model.LeafAlloc;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.perf4j.aop.Profiled;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @ClassName SegmentManager
 * @Description 分段取号管理类
 * @Author lgh
 * @Date 2020-04-08 16:09
 */
@Component
@Slf4j
public class SegmentManager {

    @Autowired
    LeafAllocMapper leafAllocMapper;

    /**
     * 是否初始成功了
     */
    private boolean initOk=false;

    /**
     * 内部缓存集合
     */
    private ConcurrentHashMap<String, SegmentBuffer> caches=new ConcurrentHashMap<>();

    /**
     * 一个segment维持时间预估为15分钟，基于该基准可对Step做扩展
     */
    private static final long SEGMENT_DURATION = 15 * 60 * 1000L;

    /**
     * 最大扩展的Id数
     */
    private static final long MAX_STEP=100000;

    /**
     * 更新时间
     */
    private long updateTimestamp=0L;


    /**
     * 线程池
     */
    private ExecutorService service = new ThreadPoolExecutor(5, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), new UpdateThreadFactory());

    /**
     * 自定义线程工厂，自定义线程名称
     */
    public static class UpdateThreadFactory implements ThreadFactory {

        private static AtomicInteger threadInitNumber = new AtomicInteger(0);

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "Thread-Segment-Update-" + threadInitNumber.incrementAndGet());
        }
    }


    /**
     * 初始化
     */
    @PostConstruct
    public void init(){
        log.info("start SegmentManager init");
        //更新bigTags
        updateLeafAllocTags();
        initOk=true;
        //开启线程自动更新数据
        updateCachesTagsEveryMin();

        log.info("end SegmentManager init");
    }


    /**
     * 根据业务标识获取ID
     */
    public long getIdByTag(String tag){

        if (!initOk){
            //尚未初始化
            log.info("init={}",initOk);
            LeafEx.LEAF_UNINIT.throwEx();
        }

        if (caches.contains(tag)){
            LeafEx.LEAF_NOT_HAS_TAG.throwEx();
        }

        //获取缓存类
        SegmentBuffer segmentBuffer=caches.get(tag);

        if (!segmentBuffer.isInit()){
            //初始化segmentBuffer
            synchronized (segmentBuffer){
                if (!segmentBuffer.isInit()){

                    log.info("init segmentBuffer={}",segmentBuffer);
                    if(!updateSegmentByDb(segmentBuffer.getCurrentSegment())){
                        LeafEx.LEAF_UPDATE_TAG_MAXID_FAIL.throwEx();
                    }

                    segmentBuffer.setInit(true);
                    log.info("init segmentBuffer={}, init ok",segmentBuffer);
                }
            }
        }
        long id=getIdFromSegmentBuffer(segmentBuffer);
        if (id<0L){
            log.info("get Id fail, tag={}",tag);
            LeafEx.LEAF_GAIN_ID_FAIL.throwEx();
        }
        return id;
    }


    /**
     * 开启线程每分钟内更新下数据内容
     */
    private void updateCachesTagsEveryMin(){
        ScheduledExecutorService scheduledExecutorService=Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName("check-idCache-thread");
                t.setDaemon(true);
                return t;
            }
        });

        scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                updateLeafAllocTags();
            }
        },1,1,TimeUnit.MINUTES);
    }


    /**
     * 更新缓存里的tags业务类别
     * @return
     */
    @Profiled(logger = "SegmentManager")
    private boolean updateLeafAllocTags(){
        //获取数据表所有配置
        List<String> bigTags=leafAllocMapper.getAllTags();
        if (bigTags==null||bigTags.size()<=0){
            return true;
        }

        //获取现存所有的旧Tags,排重
        Set<String> oldTags=caches.keySet();
        //获取新的tags，排重
        Set<String> dbTags= new HashSet<>(bigTags);

        //从缓存列表里移除失效的TAG
        for (String tag:oldTags){
            if (!dbTags.contains(tag)){
                oldTags.remove(tag);
                caches.remove(tag);
            }
        }

        // 插入新增的tag到缓存列表
        for (String tag:dbTags){

            if (!oldTags.contains(tag)){
                //初始化一个新的Segment
                SegmentBuffer segmentBuffer=new SegmentBuffer(tag);
                Segment segment=segmentBuffer.getCurrentSegment();
                segment.setBizTag(tag);
                segment.getValue().set(0);
                segment.getStep().set(0);
                segment.getMaxId().set(0);
                caches.put(tag,segmentBuffer);
                log.info("init segmentBuffer={}",segmentBuffer);
            }
        }

        log.info("caches={}",caches);
        return true;
    }


    /**
     * 根据Db更新segment
     * @param segment
     * @return
     */
    @Profiled(logger = "SegmentManager")
    private boolean updateSegmentByDb(Segment segment){

        SegmentBuffer segmentBuffer=segment.getSegmentBuffer();

        LeafAlloc leafAlloc=null;

        if (!segmentBuffer.isInit()){
            //更新节点数据
            if (leafAllocMapper.updateLeafByTag(segmentBuffer.getBigTag())<1){
                log.info("updateLeafByTag fail,tag={}:",segmentBuffer.getBigTag());
                return false;
            }
            //获取最新的内容
            leafAlloc=leafAllocMapper.getLeafByTag(segmentBuffer.getBigTag());
            segmentBuffer.setStep(leafAlloc.getStep());
            segmentBuffer.setMinStep(leafAlloc.getStep());

        } else if (updateTimestamp==0) {

            leafAlloc=leafAllocMapper.getLeafByTag(segmentBuffer.getBigTag());
            segmentBuffer.setStep(leafAlloc.getStep());
            segmentBuffer.setMinStep(leafAlloc.getStep());
            updateTimestamp=System.currentTimeMillis();

        }else {
            long timeOffest=System.currentTimeMillis()-updateTimestamp;
            int nextStep=segmentBuffer.getStep();

            if (timeOffest<SEGMENT_DURATION){

                if ((nextStep)*2<MAX_STEP){
                    //扩展
                    nextStep=nextStep*2;
                }

            }else if (timeOffest>SEGMENT_DURATION){

                //缩小step值
                nextStep=nextStep/2<segmentBuffer.getMinStep()?
                            segmentBuffer.getMinStep():nextStep/2;
            }

            //设置临时tmp
            LeafAlloc temp=new LeafAlloc();
            temp.setBizTag(segmentBuffer.getBigTag());
            temp.setStep(nextStep);
            log.info("temp={}",temp);

            if (leafAllocMapper.updateMaxIdByCustomStep(temp.getBizTag(),temp.getStep())<1){
                LeafEx.LEAF_GAIN_ID_FAIL.throwEx();
            }
            leafAlloc=leafAllocMapper.getLeafByTag(segmentBuffer.getBigTag());

            segmentBuffer.setStep(nextStep);
            segmentBuffer.setMinStep(leafAlloc.getStep());
        }

        if (leafAlloc==null){
            log.info("leafalloc gain fail tag={}",segmentBuffer.getBigTag());
            return false;
        }
        long value=leafAlloc.getMaxId()-leafAlloc.getStep();

        segment.getValue().set(value);
        segment.getStep().set(segmentBuffer.getStep());
        segment.getMaxId().set(leafAlloc.getMaxId());

        return true;
    }

    private long getIdFromSegmentBuffer(final SegmentBuffer segmentBuffer){
        while (true){
            segmentBuffer.getLock().readLock().lock();
            try{
                final Segment segment=segmentBuffer.getCurrentSegment();

                if (!segmentBuffer.isNextReady()
                        &&(segment.getMaxId().get()-segment.getValue().get())<(0.9*segment.getStep().get())
                            &&segmentBuffer.getThreadRunning().compareAndSet(false,true)){
                    service.execute(new Runnable() {
                        @Override
                        public void run() {

                            boolean isUpdateOk=false;
                            Segment next=segmentBuffer.getSegments()[segmentBuffer.nextPos()];
                            //更新
                            try{
                                log.info("updateSegmentByDb:tags={}",segment.getBizTag());
                                if(updateSegmentByDb(next)){
                                    isUpdateOk=true;
                                }
                            }catch (Exception e){
                                e.printStackTrace();
                            }finally {
                                if (isUpdateOk){
                                    segmentBuffer.getLock().writeLock().lock();
                                    segmentBuffer.setNextReady(true);
                                    segmentBuffer.getThreadRunning().set(false);
                                    segmentBuffer.getLock().writeLock().unlock();
                                }else {
                                    segmentBuffer.getThreadRunning().set(false);
                                }
                            }

                        }
                    });
                }
                long value=segment.getValue().getAndIncrement();
                if (value<segment.getMaxId().get()){
                    return value;
                }
            }finally {
                segmentBuffer.getLock().readLock().unlock();
            }

            waitAndSleep(segmentBuffer);

            segmentBuffer.getLock().writeLock().lock();
            try {
                final Segment segment=segmentBuffer.getCurrentSegment();
                long value=segment.getValue().getAndIncrement();
                if (value<segment.getMaxId().get()){
                    return value;
                }

                if (segmentBuffer.isNextReady()){
                    segmentBuffer.switchPos();
                    segmentBuffer.setNextReady(false);
                }else {
                    return -1;
                }
            }finally {
                segmentBuffer.getLock().writeLock().unlock();
            }
        }
    }

    /**
     * 自旋
     * @param buffer
     */
    private void waitAndSleep(SegmentBuffer buffer){
        int roll=0;
        while (buffer.getThreadRunning().get()){
            roll+=1;
            if (roll>10000){
                try {
                    TimeUnit.MILLISECONDS.sleep(10);
                    break;
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    break;
                }
            }
        }
    }
}
